package hust.rpc;

import hust.rpc.annotation.Service;
import hust.rpc.constants.Constants;
import hust.rpc.handler.MessageDecoder;
import hust.rpc.handler.MessageEncoder;
import hust.rpc.handler.RegistryHandler;
import hust.rpc.invoker.ConsumerRemoteInvoker;
import hust.rpc.invoker.RemoteInvoker;
import hust.rpc.lb.LoadBalance;
import hust.rpc.lb.RandomLoadBalance;
import hust.rpc.proxy.AllProxy;
import hust.rpc.proxy.Proxy;
import hust.rpc.utils.ServiceScanUtils;
import hust.rpc.utils.SingletonUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * 消费者启动器
 */
@Slf4j
public class ConsumerBootstrap {

    // 闭锁,用于同步(成功连接注册中心)
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 启动消费者连接注册中心
     *
     * @param registryIp   注册中心ip地址
     * @param registryPort 注册中心端口号
     */
    public static void run(String registryIp, int registryPort) throws InterruptedException, IOException, ClassNotFoundException {
        // 绑定服务
        bindService();
        // 设置负载均衡策略
        setLoadBalance(RandomLoadBalance.class);
        // 注册RemoteInvoker(ConsumerRemoteInvoker实现)
        // 特殊之处在于ConsumerRemoteInvoker不一定持有Channel(可能只有Host信息,需要自行创建Channel连接)
        SingletonUtils.registerInstance(RemoteInvoker.class, ConsumerRemoteInvoker.class);
        // 连接注册中心
        connectRegistry(registryIp, registryPort);
    }

    /**
     * 连接注册中心
     *
     * @param registryIp   注册中心ip地址
     * @param registryPort 注册中心端口号
     */
    private static void connectRegistry(String registryIp, int registryPort) throws InterruptedException {
        // Netty公式
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                // 关闭TCP的默认开启的Nagle算法
                .option(ChannelOption.TCP_NODELAY, true)
                // 开启TCP底层心跳机制
                .option(ChannelOption.SO_KEEPALIVE, true)
                // 设置连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new MessageEncoder());
                        pipeline.addLast(new MessageDecoder());
                        pipeline.addLast(new RegistryHandler());
                    }
                });
        log.info("服务消费者开始连接注册中心~~~");
        bootstrap.connect(registryIp, registryPort);

        // 等待同步
        countDownLatch.await();
    }

    /**
     * 设置负载均衡策略
     */
    private static void setLoadBalance(Class<? extends LoadBalance> clazz) {
        SingletonUtils.registerInstance(LoadBalance.class, clazz);
    }

    /**
     * 绑定服务
     *
     * @see hust.rpc.service.RefreshServiceImpl
     * 本地服务: 更新本地<服务, 服务提供者>映射缓存信息
     * @see hust.rpc.service.ConsumerService
     * 远程服务: 向注册中心首次拉取全量感兴趣的服务集合对应的服务提供者信息
     */
    public static void bindService() throws IOException, ClassNotFoundException {
        // 创建代理对象
        SingletonUtils.registerInstance(Proxy.class, AllProxy.class);
        // 扫描包获取@Service标注的类
        Set<Class<?>> services = ServiceScanUtils.scanService(Constants.BASE_SCAN_PACKAGE, (clazz) -> {
            Service annotation = clazz.getAnnotation(Service.class);
            // value为消费者 || value为空
            return Constants.CONSUMER.equals(annotation.value()) || annotation.value().isEmpty();
        });
        // 初始化代理对象
        SingletonUtils.getInstance(Proxy.class).initProxy(services);
    }

    /**
     * 闭锁打开
     */
    public static void countDown() {
        countDownLatch.countDown();
    }
}
